package cn.com;


import cn.com.constant.ConfigConstant;
import cn.com.core.FlinkApplication;
import org.apache.flink.api.java.utils.ParameterTool;

import java.util.Objects;

/**
 * 程序入口，用于接收各种参数，并启动任务
 *
 * @author WZQ
 * @date 2021-08-27 20:47
 **/
public class SubmitClient {

    public static void main(String[] args) {
        ParameterTool initParam = ParameterTool.fromArgs(args);
        System.out.println("输入参数：" + initParam.toMap());
        checkArgument(initParam);

        FlinkApplication flinkApplication = new FlinkApplication();
        flinkApplication.start(initParam);
    }

    /**
     * 检查输入参数是否完整、正确
     */
    private static void checkArgument(ParameterTool parameterTool) {
        if (!(parameterTool.has(ConfigConstant.EXECUTION_MODE) && ConfigConstant.EXECUTION_MODE_PATTERN.matcher(parameterTool.get(ConfigConstant.EXECUTION_MODE)).matches())) {
            throw new IllegalArgumentException("必须通过 --" + ConfigConstant.EXECUTION_MODE + " 参数，来指定sql使用批模式还是流模式执行，可用值为：streaming、batch，不区分大小写。");
        }

        if (!parameterTool.has(ConfigConstant.SQL_FILE_PATH)) {
            throw new IllegalArgumentException("必须通过 --" + ConfigConstant.SQL_FILE_PATH + " 参数，来指定sql文件的相对路径，和DS资源中的文件路径保持一致即可。");
        }

        if (Objects.equals(parameterTool.get(ConfigConstant.EXECUTION_MODE).toLowerCase(), ConfigConstant.STREAMING_MODE)) {
            // 如果是streaming流式任务，则需要查看checkpoint相关参数设置是否正确
            if (parameterTool.has(ConfigConstant.CHECKPOINT_ENABLE)) {
                if (parameterTool.getBoolean(ConfigConstant.CHECKPOINT_ENABLE)) {
                    // 开启了checkpoint
                    if (!parameterTool.has(ConfigConstant.CHECKPOINT_DIR_NAME)) {
                        throw new IllegalArgumentException("必须通过 --" + ConfigConstant.CHECKPOINT_DIR_NAME + " 参数，来指定流任务的checkpoint地址，示例：hdfs://baishancloud/data/bigdata/checkpoint/sql-");
                    }
                }
            }
        }


    }


}
